package com.apps.sdses.flink141.demo;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2023/10/9 15:54
 */
public class LookupJoinDemo {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        env.enableCheckpointing(2000);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql("CREATE TABLE orders (\n" +
                "    order_id INT,\n" +
                "    order_date TIMESTAMP(3),\n" +
                "    customer_name STRING,\n" +
                "    price DECIMAL(10, 5),\n" +
                "    product_id INT,\n" +
                "    order_status TINYINT,\n" +
                "    proctime AS PROCTIME(),\n" +
                "    PRIMARY KEY (order_id) NOT ENFORCED\n" +
                "  ) WITH (\n" +
                "    'connector' = 'mysql-cdc',\n" +
                "    'hostname' = 'localhost',\n" +
                "    'port' = '3306',\n" +
                "    'username' = 'root',\n" +
                "    'password' = '123456',\n" +
                "    'server-time-zone' = 'Asia/Shanghai',\n" +
                "    'database-name' = 'mydb',\n" +
                "    'table-name' = 'orders'\n" +
                "  )");

        tableEnv.executeSql("create TEMPORARY TABLE products(\n" +
                "   id INT, \n" +
                "   name STRING, \n" +
                "   description STRING, \n" +
                "   PRIMARY KEY (id) NOT ENFORCED\n" +
                " ) with ( \n" +
                "   'connector' = 'jdbc', \n" +
                "   'url' = 'jdbc:mysql://localhost:3306/mydb', \n" +
                "   'table-name' = 'products', \n" +
                "   'driver' = 'com.mysql.cj.jdbc.Driver', \n" +
                "   'username' = 'root', \n" +
                "   'password' = '123456' \n" +
//                "   'lookup.cache.max-rows' = '100', \n" +
//                "   'lookup.cache.ttl' = '360000' \n" +
                ")");


        tableEnv.executeSql("CREATE TABLE print_table (\n" +
                "    order_id INT,\n" +
                "    order_date TIMESTAMP(3),\n" +
                "    customer_name STRING,\n" +
                "    price DECIMAL(10, 5),\n" +
                "    product_id INT,\n" +
                "    order_status TINYINT,\n" +
                "    name STRING, \n" +
                "    description STRING \n" +
                ") WITH ('connector' = 'print')");


        //lookup join
        tableEnv.executeSql("insert into print_table " +
                " select o.order_id, o.order_date, o.customer_name, cast(11.01 as decimal(10, 2)), o.product_id, o.order_status, c.name, c.description " +
                " from orders as o " +
                " left join products FOR SYSTEM_TIME AS OF o.proctime as c " +
                " on o.product_id = c.id ");

        //常规join
//        tableEnv.executeSql("insert into print_table " +
//                " select o.order_id, o.order_date, o.customer_name, o.price, o.product_id, o.order_status, c.name, c.description " +
//                " from orders as o " +
//                " left join products as c " +
//                " on o.product_id = c.id ");




    }
}
